package abacab.remote.netty;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.remote.ConnectionRegistry;
import abacab.remote.protocol.Acceptor;

/**
 * The acceptor implementation for Netty
 */
public final class NettyAcceptor implements Acceptor {
	/** The logger */
	private static final Logger		logger	= LoggerFactory.getLogger(NettyAcceptor.class);

	/** The factory for channels */
	private final ChannelFactory	channelFactory;

	/** The channel */
	private final Channel					acceptorChannel;

	/**
	 * Create the acceptor
	 * 
	 * @param registry
	 *          the registry
	 * @throws IOException
	 *           if an error occurred
	 */
	public NettyAcceptor(final ConnectionRegistry registry) throws IOException {
		this.channelFactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool());

		ServerBootstrap bootstrap = new ServerBootstrap(this.channelFactory);
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			@Override
			public ChannelPipeline getPipeline() {
				NettyServerConnection connection = new NettyServerConnection(registry);
				ChannelPipeline pipeline = Channels.pipeline();
				pipeline.addLast("frame", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 2, 0, 2));
				pipeline.addLast("connection", connection.getHandler());
				return pipeline;
			}
		});
		bootstrap.setOption("child.tcpNoDelay", Boolean.TRUE);
		bootstrap.setOption("child.keepAlive", Boolean.TRUE);

		this.acceptorChannel = bootstrap.bind(new InetSocketAddress(registry.getNetworkInterface(), registry
				.getListeningPort()));
	}

	/**
	 * Get the local address
	 * 
	 * @return the address
	 */
	public InetSocketAddress getAddress() {
		return (InetSocketAddress) this.acceptorChannel.getLocalAddress();
	}

	/**
	 * Stop the acceptor
	 */
	@Override
	public void stop() {
		ChannelFuture future = this.acceptorChannel.close();
		if (!future.awaitUninterruptibly(5000L)) {
			logger.warn("Failed to close the acceptor in time (it may close in the future)");
		}
		this.channelFactory.releaseExternalResources();
	}
}